package com.xdf.testkafk.kafka;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xdf.testkafk.vo.KafkaMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * kafka 消息消费者类
 * @author chanchaw
 * @create 2025-10-31 12:36
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class KafkaConsumer {
    private final ObjectMapper mapper;

    /**
     * 通过注解 @KafkaListener 监听指定的消息，在本方法中消息掉
     * 已知消息记录的 KV 都是 String 时如下读取
     * @param record
     */
    @KafkaListener(topics = {"qinyi-springboot"}, groupId = "qinyi-springboot-kafka")
    public void listener01(ConsumerRecord<String,String> record) throws JsonProcessingException {
        String key = record.key();
        String value = record.value();
        KafkaMessage kafkaMessage = mapper.readValue(value, KafkaMessage.class);
        log.info("监听到 kafka 消息，key:[{}], message:[{}]", key, mapper.writeValueAsString(kafkaMessage));// 打印读取到的消息到日志
    }

    /**
     * 未知 kafka 消息的 KV 数据类型
     * 一个 groupId 只能对应一个消费者，本方法的 groupId 区别于 listener01
     * @param record
     * @throws JsonProcessingException
     */
    @KafkaListener(topics = {"qinyi-springboot"}, groupId = "qinyi-springboot-kafka-1")
    public void listener02(ConsumerRecord<?,?> record) throws JsonProcessingException {
        Optional<?> _kafkaMessage = Optional.ofNullable(record.value());
        if(!_kafkaMessage.isPresent()) return;

        Object message = _kafkaMessage.get();
        KafkaMessage kafkaMessage = mapper.readValue(message.toString(), KafkaMessage.class);
        log.info("监听到 kafka 消息, message:[{}]", mapper.writeValueAsString(kafkaMessage));// 打印读取到的消息到日志
    }
}
